package csl.blog.server.mq.consumer.service.impl;

import cn.hutool.json.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.rabbitmq.client.Channel;
import csl.blog.server.mq.consumer.entity.AsyncEventEntity;
import csl.blog.server.mapper.AsyncEventMapper;
import csl.blog.server.mq.consumer.service.AsyncEventService;
import csl.blog.server.utils.RedisUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service("asyncEventServiceImpl")
public class AsyncEventServiceImpl extends ServiceImpl<AsyncEventMapper, AsyncEventEntity> implements AsyncEventService {

    private static final String EVENT_CACHE_KEY = "ASYNC_EVENT:";

    @Autowired
    private RedisUtils redisUtils;

    @Override
    public void receiveEvent(JSONObject message, long deliveryTag, Channel channel) throws IOException {
        String cacheKey = null;
        try {
            String taskId = message.getStr("taskId");
            String taskTag = message.getStr("taskTag");
            cacheKey = EVENT_CACHE_KEY + taskId;
            //处理重复消费问题
            boolean isRepeat = handleRepeat(cacheKey, taskId, taskTag, message);
            log.info("queue:" + taskTag + " -- recieve message success!" + "message:" + message.toString());
            if (!isRepeat) {
                boolean result = handleMessage(message);
                if (result) {
                    /*签收*/
                    channel.basicAck(deliveryTag, false);
                    return;
                }
            }
            channel.basicNack(deliveryTag, false, true);
        } catch (Exception e) {
            log.error(e.getLocalizedMessage());
            channel.basicNack(deliveryTag, false, true);
        }finally {
            redisUtils.deleteObject(cacheKey);
        }
    }

    @Override
    public boolean handleRepeat(String cacheKey, String taskId, String taskTag, JSONObject content) {
        try {
            //分布式事务锁
            long incrNum = redisUtils.incrLong(cacheKey, 30, TimeUnit.SECONDS);
            if (incrNum != 0L) {
                log.warn("Duplicate async event. [taskId:{}]", taskId);
                return true;
            }
            //数据库校验
            int count = this.count(new QueryWrapper<AsyncEventEntity>().lambda().eq(AsyncEventEntity::getTaskId, taskId));
            if (count != 0) {
                log.warn("Duplicate async event. [taskId:{}]", taskId);
                return true;
            }
            //任务存储DB
            AsyncEventEntity asyncEventEntity = new AsyncEventEntity();
            asyncEventEntity.setTaskId(taskId);
            asyncEventEntity.setTaskTag(taskTag);
            asyncEventEntity.setContent(content.toString());
            asyncEventEntity.setStatus(1);
            this.save(asyncEventEntity);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public boolean handleMessage(JSONObject message) {
        String taskId = message.getStr("taskId");
        String taskTag = message.getStr("taskTag");
        String name = message.getStr("name");
        boolean result = true;
        String data = message.getStr("data");
        /*异步用户注册*/
        if ("REGISTER".equals(name)){
            log.info("收到REGISTER异步消息，开始处理: -----{}", data);
        }
        return result;
    }
}
